package com.bwie.controller;

import com.bwie.config.KafkaConfig;
import com.bwie.utils.R;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @description :
 * @Author QiHainan
 * @Create 2025/3/28 15:24
 * @Version 1.0
 */
@RestController
@RequestMapping("/user/kafka")
public class KafkaController {
    @Autowired
    KafkaTemplate kafkaTemplate;

    /**
     * 发送消息
     * @param msg
     * @return
     */
    @PostMapping("/send/{msg}")
    public R send(@PathVariable("msg") String msg){
        System.out.println("发送消息："+msg);
        kafkaTemplate.send(KafkaConfig.TOPIC_TEST,msg);
        return R.OK(msg);
    }

    @KafkaListener(topics = KafkaConfig.TOPIC_TEST)
    public void recvTestMessage(String msg, Acknowledgment acknowledgment){
        System.out.println("接收消息："+msg);

        //手动接收消息，删除消息
        acknowledgment.acknowledge();
    }
}
